/*
 * Copyright 2010 Thedwick, LLC

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.

 */
package com.thedwick.jdbcGrabber.sink;

import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import com.thedwick.jdbcGrabber.util.ClassUtil;
import com.thedwick.jdbcGrabber.util.ThreadUtil;

/**
*
* A decorator sink that wraps another sink in a size-limited queue.  When the queue
* is full, then new requests to consume() will simply be silently dropped on the floor.
* 
* This prevents a badly behaving sink or an overloaded system from getting overwhelmed with 
* data and running out of memory
* 
*/
public class LimitedQueueSink implements SQLSink {
	public static final String DEFAULT_THREADS = "1";
	public static final String DEFAULT_QUEUE_SIZE = "100";
	public static final String LIMITED_QUEUE_SINK_UNDERLYING = "LimitedQueueSink.underlying";
	public static final String LIMITED_QUEUE_SINK_THREADS = "LimitedQueueSink.threads";
	public static final String LIMITED_QUEUE_SINK_QUEUESIZE = "LimitedQueueSink.queuesize";
	private final SQLSink underlying;
	private final ExecutorService executor;
	private final BlockingQueue<TraceElement> consumeQueue;
	private int dropped;
	private final int numThreads;
	
	public LimitedQueueSink(Properties config){
		underlying = createUnderlying(config);
		numThreads = getNumThreads(config);
		executor = Executors.newFixedThreadPool(numThreads);
		consumeQueue = new LinkedBlockingQueue<TraceElement>(getQueueSize(config));
		startConsumers();
	}
	
	private void startConsumers() {
		for (int i=0; i<numThreads; i++){
			executor.execute(new QueueConsumer(underlying, consumeQueue));
		}
	}

	protected SQLSink getUnderlying(){
		return this.underlying;
	}
	
	private int getQueueSize(Properties config) {
		try {
			return Integer.parseInt(config.getProperty(LIMITED_QUEUE_SINK_QUEUESIZE, DEFAULT_QUEUE_SIZE));
		} catch (NumberFormatException e){
			throw new IllegalArgumentException(LIMITED_QUEUE_SINK_QUEUESIZE + " must be a legal integer", e);
		}
	}

	private int getNumThreads(Properties config) {
		try {
			return Integer.parseInt(config.getProperty(LIMITED_QUEUE_SINK_THREADS, DEFAULT_THREADS));
		} catch (NumberFormatException e){
			throw new IllegalArgumentException(LIMITED_QUEUE_SINK_THREADS + " must be a legal integer", e);
		}
	}

	private SQLSink createUnderlying(Properties config) {
		return (SQLSink) ClassUtil.createInstance(LIMITED_QUEUE_SINK_UNDERLYING, config, config);
	}

	@Override
	public void consume(String sql) {
		StackTraceElement[] stackTrace = ThreadUtil.getStackTrace();
		consume(new TraceElement(stackTrace, sql));
	}

	@Override
	public void consume(TraceElement trace) {
		if (!this.consumeQueue.offer(trace)){
			dropped++;//I know this isn't thread safe, but I'm not too bothered if the count is off by a few
		}
	}

	public int getDropped(){
		return dropped;
	}
	
	private class QueueConsumer implements Runnable {
		private final SQLSink underlying;
		private final BlockingQueue<TraceElement> queue;
		
		QueueConsumer(SQLSink sink, BlockingQueue<TraceElement> queue){
			this.underlying = sink;
			this.queue = queue;
		}
		
		@Override
		public void run() {
			while(true){
				TraceElement toConsume = null;
				try {
					toConsume = queue.poll(15000, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					//ignore
				}
				if (toConsume != null){
					underlying.consume(toConsume);
				}
			}
		}
		
		
	}
}